package com.item.model.consumer;

import com.item.model.util.RedisUtil;
import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.ByteArrayInputStream;
import java.io.ObjectInputStream;
import java.util.Map;
import java.util.Objects;

/**
 * 手动确认消息监听类
 * 监听了队列a
 * @author zhang-rongyao
 */
@Component
public class AckReceiver implements ChannelAwareMessageListener {

    private static Logger log = LoggerFactory.getLogger(AckReceiver.class);

    @Autowired
    RedisUtil redisUtil;

    private void processData(Map<String,String> msgMap){
        String messageId=msgMap.get("messageId");
        String messageData=msgMap.get("messageData");
        String createTime=msgMap.get("createTime");

        //redis存储messageId 防止多次消费
        if(!Objects.isNull(redisUtil.get("messageId")) && messageId.equals(String.valueOf(redisUtil.get("messageId")))) return;
        log.info("MyAckReceiver 接收messageId:[{}] messageData:[{}] createTime:[{}]",messageId,messageData,createTime);
        redisUtil.set("messageId",messageId);

    }

    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        ObjectInputStream ois =null;
        try {
            byte[] body = message.getBody();
            log.info("消费的主题消息来自[{}]",message.getMessageProperties().getConsumerQueue());
            ois = new ObjectInputStream(new ByteArrayInputStream(body));
            Map<String,String> msgMap = (Map<String,String>) ois.readObject();
            this.processData(msgMap);
            //第二个参数，手动确认可以被批处理，当该参数为 true 时，则可以一次性确认 delivery_tag 小于等于传入值的所有消息
            channel.basicAck(deliveryTag, true);
        } catch (Exception e) {
            //第二个参数，true会重新放回队列，所以需要自己根据业务逻辑判断什么时候使用拒绝
            channel.basicReject(deliveryTag, false);
            e.printStackTrace();
        }finally {
            ois.close();
        }
    }
}